package com.sunzm.hbase;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

/**
 * HBase Java API
 *
 * @author Administrator
 * @version 1.0
 * @date 2021-08-10 16:38
 */
public class HBaseJavaAPI {

    private Configuration conf = null;
    private String hbaseZookeeperQuorum = "192.168.1.158";
    private Connection connection = null;

    public void init() {

        System.out.println(HConstants.ZOOKEEPER_QUORUM + ":" + hbaseZookeeperQuorum);

        conf = HBaseConfiguration.create();

        System.out.println("初始化HBase连接.....");
        conf.set("hbase.zookeeper.quorum", hbaseZookeeperQuorum);

        try {
            connection = ConnectionFactory.createConnection(conf);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public static void main(String[] args) throws Exception {
        String tableName = "mytest";

        HBaseJavaAPI hBaseJavaAPI = new HBaseJavaAPI();

        String rowkey = "rk001";

        //hBaseJavaAPI.putTest(tableName, rowkey);
        //hBaseJavaAPI.putsTest(tableName);
        //测试中连接关闭需要很长时间，并且连接关闭后数据就没了
        hBaseJavaAPI.mutatorTest(tableName);
        //BaseJavaAPI.useTimeTest(tableName);

    }

    private void useTimeTest(String tableNameStr) throws Exception {
        init();
        TableName tableName = TableName.valueOf(tableNameStr);
        List<Put> puts1 = new ArrayList<>();

        byte[] familyBytes = Bytes.toBytes("info");
        byte[] qualifierBytes = Bytes.toBytes("name");

        for (int i = 0; i < 10000; i++) {
            Put put = new Put(Bytes.toBytes("rk300" + i));
            put.addColumn(familyBytes, qualifierBytes, Bytes.toBytes("value" + i));
            puts1.add(put);
        }

        List<Put> puts2 = new ArrayList<>();

        for (int i = 0; i < 10000; i++) {
            Put put = new Put(Bytes.toBytes("rk400" + i));
            put.addColumn(familyBytes, qualifierBytes, Bytes.toBytes("value" + i));
            puts2.add(put);
        }


        List<Put> puts3 = new ArrayList<>();

        for (int i = 0; i < 10000; i++) {
            Put put = new Put(Bytes.toBytes("rk500" + i));
            put.addColumn(familyBytes, qualifierBytes, Bytes.toBytes("value" + i));
            puts3.add(put);
        }

        //1,put
        long startTs = System.currentTimeMillis();

        Table table = connection.getTable(tableName);
        for (Put put : puts1) {
            table.put(put);
        }

        long endTs = System.currentTimeMillis();

        System.out.println("put useTime: " + (endTs - startTs));

        //2,puts
        startTs = endTs;
        table.put(puts2);
        endTs = System.currentTimeMillis();
        System.out.println("puts useTime: " + (endTs - startTs));

        //3,mutator
        BufferedMutator bufferedMutator = connection.getBufferedMutator(tableName);
        startTs = endTs;
        bufferedMutator.mutate(puts3);
        bufferedMutator.flush();
        //也可以传入一个Put对象
        //bufferedMutator.mutate(put);
        bufferedMutator.close();

        endTs = System.currentTimeMillis();
        System.out.println("mutator useTime: " + (endTs - startTs));

        close(connection);
    }


    private void mutatorTest(String tableNameStr) throws Exception {

        init();

        TableName tableName = TableName.valueOf(tableNameStr);

        //BufferedMutator bufferedMutator = connection.getBufferedMutator(tableName);

        BufferedMutatorParams params = new BufferedMutatorParams(tableName);
        //设置缓存的大小
        params.writeBufferSize(1024 * 1024);
        //开启hbase插入异常监控
        BufferedMutator.ExceptionListener listeners = new BufferedMutator.ExceptionListener() {

            @Override
            public void onException(RetriesExhaustedWithDetailsException e,
                                    BufferedMutator mutator)
                    throws RetriesExhaustedWithDetailsException {
                for (int i = 0; i < e.getNumExceptions(); i++) {
                    System.out.println("插入失败 " + e.getRow(i) + ".");
                }
            }

        };
        //开启异常监听
        params.listener(listeners);
        BufferedMutator bufferedMutator = connection.getBufferedMutator(params);
        //批量提交数据插入。
        List<Put> puts = new ArrayList<>();

        byte[] familyBytes = Bytes.toBytes("info");
        byte[] qualifierBytes = Bytes.toBytes("name");
        Put put1 = new Put(Bytes.toBytes("rk2001"), System.currentTimeMillis());
        put1.addColumn(familyBytes, qualifierBytes, Bytes.toBytes("zhangsan"));
        puts.add(put1);

        Put put2 = new Put(Bytes.toBytes("rk2002"), System.currentTimeMillis());
        put2.addColumn(familyBytes, qualifierBytes, Bytes.toBytes("lisi"));
        puts.add(put2);

        Put put3 = new Put(Bytes.toBytes("rk2003"), System.currentTimeMillis());
        put3.addColumn(familyBytes, qualifierBytes, Bytes.toBytes("wangwu"));
        puts.add(put3);

        long startTs = System.currentTimeMillis();

        bufferedMutator.mutate(puts);
        bufferedMutator.flush();
        //也可以传入一个Put对象
        //bufferedMutator.mutate(put);

        bufferedMutator.close();

        long endTs = System.currentTimeMillis();

        System.out.println("useTime: " + (endTs - startTs));

        close(connection);
    }

    private void putsTest(String tableNameStr) throws IOException {
        init();

        TableName tableName = TableName.valueOf(tableNameStr);
        Table table = connection.getTable(tableName);

        List<Put> puts = new ArrayList<>();

        byte[] familyBytes = Bytes.toBytes("info");
        byte[] qualifierBytes = Bytes.toBytes("name");
        Put put1 = new Put(Bytes.toBytes("rk1001"));
        put1.addColumn(familyBytes, qualifierBytes, Bytes.toBytes("zhangsan"));
        puts.add(put1);

        Put put2 = new Put(Bytes.toBytes("rk1002"));
        put2.addColumn(familyBytes, qualifierBytes, Bytes.toBytes("lisi"));
        puts.add(put2);

        Put put3 = new Put(Bytes.toBytes("rk1003"));
        put3.addColumn(familyBytes, qualifierBytes, Bytes.toBytes("wangwu"));
        puts.add(put3);

        long startTs = System.currentTimeMillis();

        table.put(puts);

        long endTs = System.currentTimeMillis();

        System.out.println("useTime: " + (endTs - startTs));

        close(connection);
    }

    private void putTest(String tableNameStr, String rowkey) throws IOException {
        init();

        TableName tableName = TableName.valueOf(tableNameStr);
        Table table = connection.getTable(tableName);

        Put put = new Put(Bytes.toBytes(rowkey));

        byte[] familyBytes = Bytes.toBytes("info");
        byte[] qualifierBytes = Bytes.toBytes("name");
        byte[] valueBytes = Bytes.toBytes("zhangsan");
        put.addColumn(familyBytes, qualifierBytes, valueBytes);

        long startTs = System.currentTimeMillis();
        table.put(put);
        long endTs = System.currentTimeMillis();

        System.out.println("useTime: " + (endTs - startTs));

        close(connection);
    }


    public void close(Connection connection) {
        System.out.println("关闭HBase连接.....");
        if (connection != null) {
            try {
                connection.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
}
